package com.intct.flink.cdc.stream;

import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.base.DeliveryGuarantee;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.kafka.clients.producer.ProducerConfig;

/**
 * @author gufg
 * @since 2025-07-16 11:12
 */
public class CDCTest {
    public static void main(String[] args) throws Exception {
        // 创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

//        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);

        // 配置源--source
        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                .hostname("192.168.199.185")
                .port(3306)
                .username("root")
                .password("999999")
                .databaseList("m1")
                .tableList("m1.test01")
                // 设置启动作业时，读取位置
                /*
                    initial：当没有指定 `scan.startup.mode` 时，默认取值为 `initial`，官网对
                    initial 的 [说明原文](https://nightlies.apache.org/flink/flink-cdc-docsmaster/zh/docs/connectors/cdc-connectors/mysql-cdc/#startup-reading-position) 是
                                `Performs an initial snapshot on the monitored database tables upon first
                                startup, and continue to read the latest binlog.`
                    earliest-offset：可访问的最早的 binlog 偏移量。
                    latest-offset：从 binlog 的末尾位置开始读取，这意味着只接收 connector 启动
                                    之后的变更事件流。
                    specific-offset：跳过快照阶段，从特定偏移量开始读取 binlog 事件。该偏移量可以
                                    使用二进制日志文件名和位置指定。
                    timestamp：从一个指定的时间戳位置开始读取 binlog，时间的设定很方便，但是根据
                                timestamp 定位到具体的 offset 需要经过一点儿时间。
                */
                .startupOptions(StartupOptions.latest())
                .deserializer(new JsonDebeziumDeserializationSchema()) // converts SourceRecord to JSON String
                .build();
        // 从源中获取数据据
        DataStreamSource<String> cdcCource = env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "cdc_source");

        // todo 转换

        // 配置输出
        cdcCource.print();

        // kafka输出配置
//        KafkaSink<String> kafkaSink = KafkaSink.<String>builder()
//                .setBootstrapServers("cdh-node:9092")
//                .setRecordSerializer
//                        (KafkaRecordSerializationSchema.<String>builder()
//                                .setTopic("my-intct82")
//                                .setValueSerializationSchema(new SimpleStringSchema())
//                                .build())
//                // 需要设置精确认一次语义时配置如下参数：
//                .setTransactionalIdPrefix("cdc-ds-mysql")
//                /**
//                 * KafkaSink 总共支持三种不同的语义保证（DeliveryGuarantee）。对于 DeliveryGuarantee.AT_LEAST_ONCE 和
//                 *          DeliveryGuarantee.EXACTLY_ONCE，Flink checkpoint 必须启用。
//                 *          默认情况下 KafkaSink 使用 DeliveryGuarantee.NONE。 以下是对不同语义保证的解释：
//                 * DeliveryGuarantee.NONE 不提供任何保证：消息有可能会因 Kafka broker 的原因发生丢失或因 Flink 的故障发生重复。
//                 * DeliveryGuarantee.AT_LEAST_ONCE: sink 在 checkpoint 时会等待 Kafka 缓冲区中的数据全部被 Kafka producer 确认。
//                 *          消息不会因 Kafka broker 端发生的事件而丢失，但可能会在 Flink 重启时重复，因为 Flink 会重新处理旧数据。
//                 * DeliveryGuarantee.EXACTLY_ONCE: 该模式下，Kafka sink 会将所有数据通过在 checkpoint 时提交的事务写入。
//                 *          因此，如果 consumer 只读取已提交的数据（参见 Kafka consumer 配置 isolation.level），
//                 *          在 Flink 发生重启时不会发生数据重复。然而这会使数据在 checkpoint 完成时才会可见，因此请按需调整 checkpoint 的间隔。
//                 *          请确认事务 ID 的前缀（transactionIdPrefix）对不同的应用是唯一的，以保证不同作业的事务 不会互相影响！此外，
//                 *          强烈建议将 Kafka 的事务超时时间调整至远大于 checkpoint 最大间隔 + 最大重启时间，否则 Kafka 对未提交事务的过期处理会导致数据丢失。
//                 */
//                .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
//                // 设置需要在（事务超时间 > 检查重启+检查超时间)
//                .setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, "9000")
//                .build();
//
//        // 执行输出数据到kafka
//        cdcCource.sinkTo(kafkaSink);

        // 作业启动
        env.execute();
    }
}
